package com.galeno.day10;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2720:18
 */
public class BrodcastStateDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",12345);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        //第一个流（维度流）
        //c01,收音机,insert
        //c10,图书,insert
        //c11,家电,insert
        //c12,电脑,insert
        //c12,台式电脑,update
        //c01,收音机,delete
        DataStreamSource<String> lines1 = env.socketTextStream("192.168.77.3", 9999);
        SingleOutputStreamOperator<Tuple3<String, String, String>> tpStream = lines1.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(split[0], split[1], split[2]);
            }
        });

        MapStateDescriptor<String,String> stateDescriptor=new MapStateDescriptor<String, String>("idToNameState",String.class,String.class);

        BroadcastStream<Tuple3<String, String, String>> broadcastStream = tpStream.broadcast(stateDescriptor);
        //事实数据
        //o01,c10,2000
        DataStreamSource<String> lines2 = env.socketTextStream("192.168.77.3", 11111);
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> orderStream = lines2.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(split[0], split[1], Integer.parseInt(split[2]));
            }
        });
        /**
         * 事实表与维度表进行conncet
         *
         */
        BroadcastConnectedStream<Tuple3<String, String, Integer>, Tuple3<String, String, String>> broadcastConnectedStream = orderStream.connect(broadcastStream);

        broadcastConnectedStream.process(new BroadcastProcessFunction<Tuple3<String, String, Integer>, Tuple3<String, String, String>,  Tuple4<String,String,Integer,String> >() {


           //一条维度数据调用一次该方法
            @Override
            public void processElement(Tuple3<String, String, Integer> value, ReadOnlyContext ctx, Collector<Tuple4<String, String, Integer, String>> out) throws Exception {
                String cid = value.f1;
                ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(stateDescriptor);
                String name = broadcastState.get(cid);
                out.collect(Tuple4.of(value.f0,value.f1,value.f2,name));
            }

            @Override
            public void processBroadcastElement(Tuple3<String, String, String> value, Context ctx, Collector<Tuple4<String, String, Integer, String>> out) throws Exception {
                String cid = value.f0;
                String name = value.f1;
                String type = value.f2;

                BroadcastState<String, String> broadcastState = ctx.getBroadcastState(stateDescriptor);
                if ("delete".equals(type)){
                    broadcastState.remove(cid);
                }else {
                    broadcastState.put(cid,name);
                }
            }
        }).print();
        env.execute();




    }
}
